package msg

import (
	"crypto/md5"
	"database/sql"
	"encoding/binary"
	"fmt"
	"gitee.com/tziye/renai-go/common"
	"gitee.com/tziye/renai-go/renai"
	"gitee.com/tziye/renai-go/sqlproxy"
	"strings"
	"sync/atomic"
	"time"
)

type MsgOperator struct {
	RenaiClient *renai.RenaiClient
	Db          *sql.DB
}

func NewMsgOperator(renaiClient *renai.RenaiClient) *MsgOperator {
	return &MsgOperator{
		RenaiClient: renaiClient,
		Db:          renaiClient.Db,
	}
}

const partitions = 3

var count atomic.Int32

func (msgOperator *MsgOperator) Produce(topic string, key string, value any) {
	msg := common.NewTuple2(key, value)
	msgOperator.commonProduce(topic, []common.Tuple2{msg})
}

func (msgOperator *MsgOperator) BatchProduce(topic string, msgs []common.Tuple2) {
	msgOperator.commonProduce(topic, msgs)
}

func (msgOperator *MsgOperator) commonProduce(topic string, msgs []common.Tuple2) {
	s := "INSERT INTO renai_msg_produce(topic, pt, k, msg, g, producer) VALUES('%s', %d, '%s', '%s', '%s', '%s')"
	sqlList := make([]string, 0)
	for _, msg := range msgs {
		group := msgOperator.RenaiClient.Instance.G + common.DELIMITER + msgOperator.RenaiClient.Instance.Service
		producer := msgOperator.RenaiClient.Instance.InstanceId
		k := common.String(msg.V1)
		ss := fmt.Sprintf(s, topic, calPartition(k), k, common.String(msg.V2), group, producer)
		sqlList = append(sqlList, ss)
	}
	sqlproxy.BatchExec(msgOperator.Db, sqlList)
}

func calPartition(key string) int {
	if strings.TrimSpace(key) != "" {
		bs := md5.Sum([]byte(key))
		return int(binary.BigEndian.Uint64(bs[0:]) % partitions)
	}
	return int(count.Add(1) % partitions)
}

func (msgOperator *MsgOperator) Consume(topic string, group string, count int, start time.Time) []RenaiMsg {
	rows := sqlproxy.CallRows(msgOperator.Db, common.MSG_CONSUME, topic, group, count, start,
		msgOperator.RenaiClient.Instance.InstanceId)
	defer rows.Close()
	msgs := make([]RenaiMsg, 0)
	for rows.Next() {
		var msg RenaiMsg
		err := rows.Scan(&msg.Id, &msg.Topic, &msg.Pt, &msg.K, &msg.Msg, &msg.CreateTime)
		common.PanicErr(err)
		msgs = append(msgs, msg)
	}
	return msgs
}

func (msgOperator *MsgOperator) Ack(topic string, pt int, group string, msgId int64) {
	s := `UPDATE renai_msg_consume SET consumed_id = ? 
          	WHERE topic = ? AND pt = ? AND g = ? AND consumer = ? 
          	AND consumed_id <= ? AND occupied_id >= ?`
	sqlproxy.Exec(msgOperator.Db, s, msgId, topic, pt, group, msgOperator.RenaiClient.Instance.InstanceId, msgId, msgId)
}

func (msgOperator *MsgOperator) Replay(topic string, startTime time.Time, endTime time.Time, count int, curId int64) []RenaiMsg {
	s := `SELECT * FROM renai_msg_produce
			WHERE topic = ? AND id > ? AND create_time >= ? AND create_time <= ?
			ORDER BY id LIMIT ?`
	rows := sqlproxy.QueryRows(msgOperator.Db, s, topic, curId, startTime, endTime, count)
	return fetchMsg(rows)
}

func (msgOperator *MsgOperator) ReplayById(topic string, minId int64, maxId int64, count int, curId int64) []RenaiMsg {
	s := `SELECT * FROM renai_msg_produce
			WHERE topic = ? AND id > ? AND id >= ? AND id <= ?
			ORDER BY id LIMIT ?`
	rows := sqlproxy.QueryRows(msgOperator.Db, s, topic, curId, minId, maxId, count)
	return fetchMsg(rows)
}

func fetchMsg(rows *sql.Rows) []RenaiMsg {
	defer rows.Close()
	var msgs []RenaiMsg
	for rows.Next() {
		var msg RenaiMsg
		err := rows.Scan(&msg.Id, &msg.Topic, &msg.Pt, &msg.K, &msg.Msg, &msg.G, &msg.Producer, &msg.CreateTime)
		common.PanicErr(err)
		msgs = append(msgs, msg)
	}
	return msgs
}

func (msgOperator *MsgOperator) Delete(topic string) {
	s1 := fmt.Sprintf("DELETE FROM renai_msg_consume WHERE topic = '%s'", topic)
	s2 := fmt.Sprintf("DELETE FROM renai_msg_archive WHERE topic = '%s'", topic)
	s3 := fmt.Sprintf("DELETE FROM renai_msg_produce_cal WHERE topic ='%s'", topic)
	s4 := fmt.Sprintf("DELETE FROM renai_msg_produce WHERE topic = '%s'", topic)
	sqlproxy.BatchExec(msgOperator.Db, []string{s1, s2, s3, s4})
}

type RenaiMsg struct {
	Id         int64
	Topic      string
	Pt         int
	K          string
	Msg        string
	G          string
	Producer   string
	CreateTime time.Time
}

// RenaiTopic

type RenaiTopic struct {
	RenaiInstance renai.RenaiInstance
	MsgOperator   *MsgOperator
	Topic         string
	Group         string
	Count         int
	Interval      int
	ReplayCount   int
}

func NewRenaiTopicGroup(renaiClient *renai.RenaiClient, topic string, group string) *RenaiTopic {
	return &RenaiTopic{
		RenaiInstance: renaiClient.Instance,
		MsgOperator:   NewMsgOperator(renaiClient),
		Topic:         topic,
		Group:         group,
		Count:         100,
		Interval:      100,
		ReplayCount:   500,
	}
}

func NewRenaiTopic(renaiClient *renai.RenaiClient, topic string) *RenaiTopic {
	defaultGroup := renaiClient.Instance.G + common.DELIMITER + renaiClient.Instance.Service
	return NewRenaiTopicGroup(renaiClient, topic, defaultGroup)
}

func (renaiTopic *RenaiTopic) Produce(value any) {
	renaiTopic.MsgOperator.Produce(renaiTopic.Topic, "", value)
}

func (renaiTopic *RenaiTopic) BatchProduce(values []any) {
	msgs := make([]common.Tuple2, 0)
	for _, value := range values {
		msg := common.NewTuple2("", value)
		msgs = append(msgs, msg)
	}
	renaiTopic.MsgOperator.BatchProduce(renaiTopic.Topic, msgs)
}

func (renaiTopic *RenaiTopic) ProduceWithKey(key string, value any) {
	renaiTopic.MsgOperator.Produce(renaiTopic.Topic, key, value)
}

func (renaiTopic *RenaiTopic) BatchProduceWithKey(msgs []common.Tuple2) {
	renaiTopic.MsgOperator.BatchProduce(renaiTopic.Topic, msgs)
}

func (renaiTopic *RenaiTopic) Consume() []RenaiMsg {
	return renaiTopic.MsgOperator.Consume(renaiTopic.Topic, renaiTopic.Group, renaiTopic.Count, time.Now())
}

func (renaiTopic *RenaiTopic) ConsumeStart(startTime time.Time) []RenaiMsg {
	return renaiTopic.MsgOperator.Consume(renaiTopic.Topic, renaiTopic.Group, renaiTopic.Count, startTime)
}

func (renaiTopic *RenaiTopic) Ack(msg RenaiMsg) {
	renaiTopic.MsgOperator.Ack(renaiTopic.Topic, msg.Pt, renaiTopic.Group, msg.Id)
}

type Consumer func(msg string, params ...any) error

func (renaiTopic *RenaiTopic) AutoConsume(consumer Consumer, params ...any) {
	renaiTopic.AutoConsumeStart(time.Now(), consumer, params...)
}

func (renaiTopic *RenaiTopic) AutoConsumeStart(startTime time.Time, consumer Consumer, params ...any) {
	for {
		var lastMsg RenaiMsg
		msgList := renaiTopic.ConsumeStart(startTime)
		for _, msg := range msgList {
			err := consumer(msg.Msg, params...)
			common.LogErr(err)
			if err != nil {
				break
			}
			lastMsg = msg
		}
		if lastMsg.Id != 0 {
			renaiTopic.Ack(lastMsg)
			common.SleepMS(renaiTopic.Interval)
		} else {
			common.SleepMS(renaiTopic.Interval * 5)
		}
	}
}

func (renaiTopic *RenaiTopic) Replay(startTime time.Time, endTime time.Time, consumer Consumer, params ...any) {
	var curId int64 = 0
	size := renaiTopic.ReplayCount
	for size == renaiTopic.ReplayCount {
		msgs := renaiTopic.MsgOperator.Replay(renaiTopic.Topic, startTime, endTime, size, curId)
		for _, msg := range msgs {
			err := consumer(msg.Msg, params...)
			common.PanicErr(err)
			curId = msg.Id
		}
		size = len(msgs)
	}
}

func (renaiTopic *RenaiTopic) ReplayById(minId int64, maxId int64, consumer Consumer, params ...any) {
	var curId int64 = 0
	size := renaiTopic.ReplayCount
	for size == renaiTopic.ReplayCount {
		msgs := renaiTopic.MsgOperator.ReplayById(renaiTopic.Topic, minId, maxId, size, curId)
		for _, msg := range msgs {
			err := consumer(msg.Msg, params...)
			common.PanicErr(err)
			curId = msg.Id
		}
		size = len(msgs)
	}
}
